package org.zjt.demo;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.zjt.demo.util.HBaseHelper;

/**
 * DESC
 *
 * @author
 * @create 2017-04-18 下午4:33
 **/
public class HbaseTest {

    private static Connection connection;
    private static String tableName = "testtable";
    private static String colfam = "colfam1";

    static {
        try {
            connection = getConnetcion(tableName, colfam);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void main(String[] args) throws Exception {
        testCAM();
    }


    public static Connection getConnetcion(String table, String colfam) throws Exception {
        Configuration conf = HBaseConfiguration.create(); // co GetExample-1-CreateConf Create the configuration.
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        // ^^ GetExample
        HBaseHelper helper = HBaseHelper.getHelper(conf);
        if (!helper.existsTable(table)) {
            helper.createTable(table, colfam);
        }
        // vv GetExample
        return ConnectionFactory.createConnection(conf);
    }


    public static void closeConnetcion(Connection connection) throws Exception {
        if (connection != null)
            connection.close();
    }


    /**
     * 测试Htable执行单次get ：单次get
     *
     * @throws Exception
     */
    public static void testGet() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes("row1"));
        get.addColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"));
        get.addColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("age"));
        get.setMaxVersions();
        Result result = table.get(get);

        byte[] val = result.getValue(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"));
        byte[] age = result.getValue(Bytes.toBytes("colfam1"), Bytes.toBytes("age"));
        System.out.println(String.format("qual1:%s\t age:%s", Bytes.toString(val), Bytes.toString(age)));
        for (Cell cell : result.rawCells()) {
            System.out.println("Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
        }
        table.close();
        closeConnetcion(connection);
    }

    /**
     * 测试Htable执行多次get：多次get
     *
     * @throws Exception
     */
    public static void testGets() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));
        List<Get> gets = new ArrayList<>();
        String[] quals = {"qual1", "age", "qq"};
        Arrays.stream(quals).forEach(a -> {
            Get get = new Get(Bytes.toBytes("row1"));
            get.addColumn(Bytes.toBytes("colfam1"), Bytes.toBytes(a));
            gets.add(get);
        });
        Result[] resultArray = table.get(gets);
        for (Result result : resultArray)
            for (Cell cell : result.rawCells()) {
                System.out.println("Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
            }
        table.close();
        closeConnetcion(connection);
    }

    /**
     * batch实现批量的处理。
     * PS: GET PUT DELETE 的row的子类
     *
     * @throws Exception
     */
    public static void testBatch() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));
        List<Row> rows = new ArrayList<>();

        // TODO: 2017/4/19 get多操作
        String[] quals = {"qual1", "age", "qq"};
        Arrays.stream(quals).forEach(a -> {
            Get get = new Get(Bytes.toBytes("row1"));
            get.addColumn(Bytes.toBytes("colfam1"), Bytes.toBytes(a));
            rows.add(get);
        });

        /** error: scan不是row的子类 */
        // TODO: 2017/4/19 delete删除操作 
        Delete delete = new Delete(Bytes.toBytes("row2"), 2);
        rows.add(delete);

        // TODO: 2017/4/19 put新增操作
        Put put = new Put(Bytes.toBytes("row2"));
        put.add(new KeyValue(Bytes.toBytes("row2"), Bytes.toBytes(colfam), Bytes.toBytes("content"), Bytes.toBytes("content")));
        put.setTTL(4000000L);
        rows.add(put);

        Object[] resultArray = new Object[rows.size()];
        /** 异常也能得到返回结果 */
        table.batch(rows, resultArray);
        Arrays.stream(resultArray).forEach(a -> {
            // TODO: 2017/4/19 执行get得到的结果处理
            if (a instanceof Result) {
                Result var = (Result) a;
                System.out.println(a);
                Arrays.stream(var.rawCells()).map(cell -> "Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())).forEach(System.out::println);
            } else
                System.out.println(a);
        });
        table.close();
        closeConnetcion(connection);
    }


    /**
     * scan扫描 （带分页）
     *
     * @throws Exception
     */
    public static void testScan() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(colfam));
        /**
         * 可添加多个column family：
         * scan.addFamily(Bytes.toBytes(colfam));
         * */

        scan.setStartRow(Bytes.toBytes("row2"));
        scan.setMaxResultSize(0);
        Filter filter = new PageFilter(4);
        scan.setFilter(filter);
        ResultScanner scanner = table.getScanner(scan);
        scanner.forEach(a -> Arrays.stream(a.rawCells()).map(cell -> "Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())).forEach(System.out::println));
    }


    /**
     * CAS 实现 无该数据插入，预设值更新
     *
     * @throws Exception
     */
    public static void testCAS() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));

        /**
         * 1、插入CAS：当该数据不存在时候，添加
         */
        Put put = new Put(Bytes.toBytes("row3"));
        put.add(new KeyValue(Bytes.toBytes("row3"), Bytes.toBytes(colfam), Bytes.toBytes("content"), Bytes.toBytes("content")));
        /** 老值value为null,表示不存在该数据  新数据则添加*/
        table.checkAndPut(Bytes.toBytes("row3"), Bytes.toBytes(colfam), Bytes.toBytes("content"), null, put);

        /**
         * 2、更新CAS：当该数据等于预设值的时候，更改。
         */
        Put put1 = new Put(Bytes.toBytes("row3"));
        put1.add(new KeyValue(Bytes.toBytes("row3"), Bytes.toBytes(colfam), Bytes.toBytes("content"), Bytes.toBytes("update-content")));
        /** 老的值为content才会更新 */
        table.checkAndPut(Bytes.toBytes("row3"), Bytes.toBytes(colfam), Bytes.toBytes("content"), Bytes.toBytes("content"), put1);

        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(colfam));
        scan.setStartRow(Bytes.toBytes("row3"));
        scan.setMaxResultSize(0);
        Filter filter = new PageFilter(4);
        scan.setFilter(filter);
        ResultScanner scanner = table.getScanner(scan);
        scanner.forEach(a -> Arrays.stream(a.rawCells()).map(cell -> "Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())).forEach(System.out::println));
        table.close();
        closeConnetcion(connection);
    }


    /**
     * CAD 比较删除，比较设置的值等于老值，就删除
     *
     * @throws Exception
     */
    public static void testCAD() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes("row4"));
        delete.addFamily(Bytes.toBytes(colfam));
        delete.addColumn(Bytes.toBytes(colfam), Bytes.toBytes("age"));
        table.checkAndDelete(Bytes.toBytes("row4"), Bytes.toBytes(colfam), Bytes.toBytes("age"), Bytes.toBytes("18"), delete);
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(colfam));
        scan.setStartRow(Bytes.toBytes("row3"));
        scan.setMaxResultSize(0);
        Filter filter = new PageFilter(4);
        scan.setFilter(filter);
        ResultScanner scanner = table.getScanner(scan);
        scanner.forEach(a -> Arrays.stream(a.rawCells()).map(cell -> "Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())).forEach(System.out::println));
        table.close();
        closeConnetcion(connection);
    }

    /**
     * CAM 同一行的多次操作
     * @throws Exception
     */
    public static void testCAM() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));
        RowMutations rowMutations = new RowMutations(Bytes.toBytes("row4"));

        Delete delete = new Delete(Bytes.toBytes("row4"));
        delete.addFamily(Bytes.toBytes(colfam));
        delete.addColumn(Bytes.toBytes(colfam), Bytes.toBytes("age"));
        rowMutations.add(delete);

        Put put = new Put(Bytes.toBytes("row4"));
        put.add(new KeyValue(Bytes.toBytes("row4"), Bytes.toBytes(colfam), Bytes.toBytes("content"), Bytes.toBytes("content")));
        put.setTTL(4000000L);
        rowMutations.add(put);

        boolean res1 = table.checkAndMutate(Bytes.toBytes("row4"), Bytes.toBytes(colfam), Bytes.toBytes("age"), CompareFilter.CompareOp.EQUAL, Bytes.toBytes("18"), rowMutations);

        System.out.println("res1="+res1);
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(colfam));
        scan.setStartRow(Bytes.toBytes("row3"));
        scan.setMaxResultSize(0);
        Filter filter = new PageFilter(4);
        scan.setFilter(filter);
        ResultScanner scanner = table.getScanner(scan);
        scanner.forEach(a -> Arrays.stream(a.rawCells()).map(cell -> "Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())).forEach(System.out::println));
        table.close();
        closeConnetcion(connection);
    }


}
